package com.txl.cn.spark04


import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
  * Created by txl on 2018/1/2.
  */
object Impclick extends  App{
  //1010,华语剧场|剧情|当代|类型,1,0
  val conf = new SparkConf().setAppName("imp").setMaster("local")
  val sc=new SparkContext(conf)
  val logs = sc.textFile("data/impclick.txt")
  val res: RDD[((String, String), Int, Int)] = logs.flatMap({
    log =>
      val str = log.split(",")
      val id = str(0)
      val typ = str(1)
      val imp = str(2).toInt
      val click = str(3).toInt
      val types: Array[String] = typ.split("[|]")
      val data= types.map({
        t =>
          ((id, t), imp, click)
      })
      data
  })

   val res2 = res.groupBy(_._1)
   val res3: RDD[((String, String), (Int, Int))] = res2.mapValues({
    it =>
      val totalimp = it.map({
        x =>
          x._2
      }).sum

      val totalclick = it.map({
        x =>
          x._3
      }).sum
      (totalimp, totalclick)
  })
  val res4=res3.coalesce(1).sortBy(t=>(t._1._1,-t._2._1,-t._2._2))
  //val res4=res3.collect.sortBy(t=>(t._1._1,-t._2._1,-t._2._2))
  res4.foreach(println)






}
